package com.gitee.wsl.collections.iterator.api


import kotlin.collections.iterator
import kotlin.experimental.ExperimentalTypeInference
import kotlin.jvm.JvmName

/**
 * An extended iterator with resource closing functionality (extends [AutoCloseable]).
 *
 * It is not [Sequence] but [Iterator], since we can't control every method of [Sequence].
 *
 * The implementation must [close] the underling resource when there are no more items as well.
 * This will make all the _terminal_ operations (e.g. [toList]) and
 * the _intermediate_ operations described by this interface (e.g. [forEach]) safe to use,
 * which means no explicit closing is required (but desirable).
 * Closing the iterator at its end will also make certain methods
 * of the child sequence ([Iterator.asSequence]) safe.
 * But **not** all the methods: such methods as [Sequence.take] is not terminal,
 * so the source resource iterator must be closed explicitly.
 * Or instead of [Iterator.asSequence] the method [asSafeSequence] could be used.
 *
 * In any case, the explicit iterator closing is highly recommended since any operations may throw an exceptions.
 */
@Suppress("INAPPLICABLE_JVM_NAME")
@OptIn(ExperimentalTypeInference::class, ExperimentalStdlibApi::class)
interface ResourceIterator<out X> : Iterator<X>, AutoCloseable {

    /**
     * Returns a resource iterator containing only elements matching the given [predicate].
     * The operation is _intermediate_ and _stateless_.
     */
    fun filter(predicate: (X) -> Boolean): ResourceIterator<X>

    /**
     * Returns a sequence containing first elements satisfying the given [predicate].
     * The operation is _intermediate_ and _stateless_.
     */
    fun takeWhile(predicate: (X) -> Boolean): ResourceIterator<X>

    /**
     * Returns a resource iterator containing the results of applying the given [transform] function
     * to each element in this iterator.
     * The operation is _intermediate_ and _stateless_.
     */
    fun <R> map(transform: (X) -> R): ResourceIterator<R>

    /**
     * Returns a single resource iterator of all elements from results of [transform] function
     * being invoked on each element of this iterator.
     * The operation is _intermediate_ and _stateless_.
     */
    @OverloadResolutionByLambdaReturnType
    @JvmName("flatMapIterable")
    fun <R> flatMap(transform: (X) -> Iterable<R>): ResourceIterator<R>

    /**
     * Returns a single resource iterator of all elements from results of [transform] function
     * being invoked on each element of this iterator.
     * The operation is _intermediate_ and _stateless_.
     */
    @OverloadResolutionByLambdaReturnType
    @JvmName("flatMapIterator")
    fun <R> flatMap(transform: (X) -> Iterator<R>): ResourceIterator<R>

    /**
     * Returns a resource iterator containing only distinct elements from this iterator.
     * Among equal elements of the iterator, only the first one will be present in the resulting iterator.
     * The elements in the resulting iterator are in the same order as they were in the source.
     * The operation is _intermediate_ and _stateful_.
     */
    fun distinct(): ResourceIterator<X>

    /**
     * Returns a resource iterator containing all elements of this iterator
     * and then all elements of the given "[other]" iterator.
     * The operation is _intermediate_ and _stateless_.
     */
    operator fun plus(other: Iterator<@UnsafeVariance X>): ResourceIterator<X>

    /**
     * Returns the first element.
     * The operation is _terminal_.
     * @throws NoSuchElementException if the iterator is empty
     */
    fun first(): X {
        use {
            if (!hasNext())
                throw NoSuchElementException("Iterator is empty.")
            return next()
        }
    }

    /**
     * Returns the first element, or `null` if the iterator is empty.
     * The operation is _terminal_.
     */
    fun firstOrNull(): X? {
        use {
            if (!hasNext())
                return null
            return next()
        }
    }

    /**
     * Returns the single element, or throws an exception if the iterator is empty or has more than one element.
     * The operation is _terminal_.
     */
    fun single(): X {
        use {
            if (!hasNext())
                throw NoSuchElementException("Iterator is empty.")
            val single = next()
            if (hasNext())
                throw IllegalArgumentException("Iterator has more than one element.")
            return single
        }
    }

    /**
     * Returns the single element, or `null` if the iterator is empty or has more than one element.
     * The operation is _terminal_.
     */
    fun singleOrNull(): X? {
        use {
            if (!hasNext())
                return null
            val single = next()
            if (hasNext())
                return null
            return single
        }
    }

    /**
     * Returns the number of elements in this iterator.
     * The operation is _terminal_.
     */
    fun count(): Long {
        use {
            var res = 0L
            while (hasNext()) {
                next()
                res++
            }
            return res
        }
    }

    /**
     * Returns a new [MutableList] filled with all elements of this iterator.
     * The operation is _terminal_.
     */
    fun toMutableList(): MutableList<@UnsafeVariance X> {
        use {
            val res = mutableListOf<X>()
            while (hasNext()) {
                res.add(next())
            }
            return res
        }
    }

    /**
     * Returns a [List] containing all elements.
     * The operation is _terminal_.
     */
    fun toList(): List<X> {
        use {
            return toMutableList()
        }
    }

    /**
     * Returns a [Set] of all elements.
     * The returned set preserves the element iteration order of the original iterator.
     * The operation is _terminal_.
     */
    fun toSet(): Set<X> {
        use {
            val res = mutableSetOf<X>()
            while (hasNext()) {
                res.add(next())
            }
            return res
        }
    }

    /**
     * Creates a sequence that returns all elements from this iterator.
     * The sequence is constrained to be iterated only once.
     * It is safe: the resource iterator is closed on method returning,
     * subsequent operations do not require resource closure anymore.
     * The operation is _terminal_.
     */
    fun asSafeSequence(): Sequence<X> {
        use {
            // can't control sequence, so collect everything in mem and close iterator
            return toList().asSequence()
        }
    }

}


@Suppress("INAPPLICABLE_JVM_NAME")
@OptIn(ExperimentalTypeInference::class)
internal abstract class BaseResourceIterator<X>(
    private val onClose: () -> Unit,
) : ResourceIterator<X> {
    private var open: Boolean = true

    override fun filter(predicate: (X) -> Boolean): ResourceIterator<X> {
        checkOpen()
        return FilteringResourceIterator(this, onClose, predicate)
    }

    override fun takeWhile(predicate: (X) -> Boolean): ResourceIterator<X> {
        checkOpen()
        return TakeWhileResourceIterator(this, onClose, predicate)
    }

    override fun <R> map(transform: (X) -> R): ResourceIterator<R> {
        checkOpen()
        return TransformingResourceIterator(this, { close() }, transform)
    }

    @OverloadResolutionByLambdaReturnType
    @JvmName("flatMapIterable")
    override fun <R> flatMap(transform: (X) -> Iterable<R>): ResourceIterator<R> {
        checkOpen()
        return FlatteningResourceIterator(this, transform, Iterable<R>::iterator)
    }

    @OverloadResolutionByLambdaReturnType
    @JvmName("flatMapIterator")
    override fun <R> flatMap(transform: (X) -> Iterator<R>): ResourceIterator<R> {
        checkOpen()
        return FlatteningResourceIterator(source = this, transformer = transform, iteratorMap = { it })
    }

    override fun distinct(): ResourceIterator<X> {
        val seen = hashSetOf<X>()
        return filter { seen.add(it) }
    }

    override fun plus(other: Iterator<X>): ResourceIterator<X> {
        return CompoundResourceIterator(this, other)
    }

    override fun close() {
        if (open) {
            open = false
            onClose()
        }
    }

    protected fun open(): Boolean = open

    protected fun checkOpen() {
        check(open) { "Iterator is closed." }
    }

    internal fun asInternalSequence(): Sequence<X> {
        return Sequence { this }.constrainOnce()
    }
}

internal open class WrappedResourceIterator<X>(
    protected val source: Iterator<X>,
    private val onClose: () -> Unit,
) : BaseResourceIterator<X>({
    safeExec(
        { (source as? AutoCloseable)?.close() },
        { onClose() },
    )
}) {

    internal fun withOnClose(onClose: () -> Unit): WrappedResourceIterator<X> {
        return WrappedResourceIterator(this.source, onClose)
    }

    override fun hasNext(): Boolean {
        if (!open()) {
            return false
        }
        if (source.hasNext()) {
            return true
        }
        close()
        return false
    }

    override fun next(): X {
        checkOpen()
        try {
            return source.next()
        } catch (ex: NoSuchElementException) {
            close()
            throw ex
        }
    }
}

/**
 * see `kotlin.sequences.FilteringSequence`
 */
internal class FilteringResourceIterator<X>(
    source: Iterator<X>,
    onClose: () -> Unit,
    private val predicate: (X) -> Boolean
) : WrappedResourceIterator<X>(source = source, onClose = onClose) {

    private var nextItem: X? = null

    // -1 for next unknown, 0 for done, 1 for continue
    private var nextState: Int = -1

    private fun calcNext() {
        while (source.hasNext()) {
            val item = source.next()
            if (predicate(item)) {
                nextItem = item
                nextState = 1
                return
            }
        }
        nextState = 0
        close()
    }

    override fun next(): X {
        checkOpen()
        if (nextState == -1)
            calcNext()
        if (nextState == 0)
            throw NoSuchElementException()
        val result = nextItem
        nextItem = null
        nextState = -1
        @Suppress("UNCHECKED_CAST")
        return result as X
    }

    override fun hasNext(): Boolean {
        if (nextState == -1)
            calcNext()
        return nextState == 1
    }
}

/**
 * see `kotlin.sequences.TakeWhileSequence`
 */
internal class TakeWhileResourceIterator<X>(
    source: Iterator<X>,
    onClose: () -> Unit,
    private val predicate: (X) -> Boolean
) : WrappedResourceIterator<X>(source = source, onClose = onClose) {

    private var nextItem: X? = null

    // -1 for next unknown, 0 for done, 1 for continue
    private var nextState: Int = -1

    private fun calcNext() {
        if (source.hasNext()) {
            val item = source.next()
            if (predicate(item)) {
                nextItem = item
                nextState = 1
                return
            }
        }
        nextState = 0
        close()
    }

    override fun next(): X {
        checkOpen()
        if (nextState == -1)
            calcNext()
        if (nextState == 0)
            throw NoSuchElementException()
        val result = nextItem
        nextItem = null
        nextState = -1
        @Suppress("UNCHECKED_CAST")
        return result as X
    }

    override fun hasNext(): Boolean {
        if (nextState == -1)
            calcNext()
        return nextState == 1
    }
}

/**
 * see `kotlin.sequences.TransformingSequence`
 */
internal class TransformingResourceIterator<T, R>(
    private val source: Iterator<T>,
    onClose: () -> Unit,
    private val transformer: (T) -> (R),
) : BaseResourceIterator<R>({
    safeExec(
        { (source as? AutoCloseable)?.close() },
        { onClose() },
    )
}) {

    override fun hasNext(): Boolean {
        if (!open()) {
            return false
        }
        if (source.hasNext()) {
            return true
        }
        close()
        return false
    }

    override fun next(): R {
        checkOpen()
        try {
            return transformer(source.next())
        } catch (ex: NoSuchElementException) {
            close()
            throw ex
        }
    }
}

/**
 * see also `kotlin.sequences.GeneratorSequence`
 */
internal class GeneratorResourceIterator<T>(
    private val getNext: () -> T?,
    onClose: () -> Unit,
) : BaseResourceIterator<T>(onClose) {

    private var nextItem: T? = null

    // -1 for next unknown, 0 for done, 1 for continue
    private var nextState: Int = -1

    private fun calcNext() {
        nextItem = getNext()
        nextState = if (nextItem == null) 0 else 1
        if (nextState == 0) {
            close()
        }
    }

    @Suppress("UNCHECKED_CAST")
    override fun next(): T {
        checkOpen()
        if (nextState < 0) {
            calcNext()
        }
        if (nextState == 0) {
            throw NoSuchElementException()
        }
        val res = nextItem as T
        nextState = -1
        return res
    }

    override fun hasNext(): Boolean {
        if (!open()) {
            return false
        }
        if (nextState < 0) {
            calcNext()
        }
        return nextState == 1
    }
}

/**
 * see `kotlin.sequences.FlatteningSequence`
 */
internal class FlatteningResourceIterator<T, R, E>(
    private val source: ResourceIterator<T>,
    private val transformer: (T) -> R,
    private val iteratorMap: (R) -> Iterator<E>,
    private var openIterators: MutableList<Iterator<*>> = mutableListOf<Iterator<*>>().also { it.add(source) }
) : BaseResourceIterator<E>({
    openIterators.closeAll()
    openIterators.clear()
}) {
    private var itemIterator: Iterator<E>? = null

    override fun next(): E {
        checkOpen()
        if (!ensureItemIterator())
            throw NoSuchElementException()
        return checkNotNull(itemIterator).next()
    }

    override fun hasNext(): Boolean {
        return ensureItemIterator()
    }

    private fun ensureItemIterator(): Boolean {
        if (!open()) {
            return false
        }
        if (itemIterator?.hasNext() == false) {
            itemIterator = null
        }
        while (itemIterator == null) {
            if (!source.hasNext()) {
                close()
                return false
            } else {
                val element = source.next()
                val nextItemIterator = iteratorMap(transformer(element))
                if (nextItemIterator.hasNext()) {
                    itemIterator = nextItemIterator
                    openIterators.add(itemIterator as Iterator<*>)
                    return true
                }
            }
        }
        return true
    }
}

internal class CompoundResourceIterator<T>(
    left: Iterator<T>,
    right: Iterator<T>,
    private var pending: MutableList<Iterator<T>?> = mutableListOf(),
    private var handled: MutableSet<Iterator<T>> = mutableSetOf(),
) : BaseResourceIterator<T>({
    safeExec(
        { handled.closeAll() },
        { pending.closeAll() },
    )
    pending.clear()
    handled.clear()
}) {
    private var index = 0
    private var current: Iterator<T> = left

    init {
        pending.add(right)
        handled.add(current)
    }

    override fun hasNext(): Boolean {
        if (!open()) {
            return false
        }
        while (!current.hasNext() && index < pending.size) {
            current = advance()
        }
        if (current.hasNext()) {
            return true
        }
        close()
        return false
    }

    override fun next(): T {
        checkOpen()
        if (!hasNext()) {
            throw NoSuchElementException()
        }
        return current.next()
    }

    override fun plus(other: Iterator<T>): ResourceIterator<T> {
        pending.add(other)
        return this
    }

    private fun advance(): Iterator<T> {
        val res = checkNotNull(pending[index])
        handled.add(res)
        pending[index] = null
        index += 1
        return res
    }
}

@Suppress("INAPPLICABLE_JVM_NAME")
@OptIn(ExperimentalTypeInference::class)
internal class EmptyResourceIterator<X> : BaseResourceIterator<X>({}) {

    override fun filter(predicate: (X) -> Boolean): ResourceIterator<X> {
        return EmptyResourceIterator()
    }

    override fun takeWhile(predicate: (X) -> Boolean): ResourceIterator<X> {
        return EmptyResourceIterator()
    }

    override fun <R> map(transform: (X) -> R): ResourceIterator<R> {
        return EmptyResourceIterator()
    }

    @OverloadResolutionByLambdaReturnType
    @JvmName("flatMapIterable")
    override fun <R> flatMap(transform: (X) -> Iterable<R>): ResourceIterator<R> {
        return EmptyResourceIterator()
    }

    @OverloadResolutionByLambdaReturnType
    @JvmName("flatMapIterator")
    override fun <R> flatMap(transform: (X) -> Iterator<R>): ResourceIterator<R> {
        return EmptyResourceIterator()
    }

    override fun distinct(): ResourceIterator<X> {
        return EmptyResourceIterator()
    }

    override fun close() {
    }

    override fun plus(other: Iterator<X>): ResourceIterator<X> {
        return if (other is AutoCloseable) {
            other.asResourceIterator { other.close() }
        } else {
            other.asResourceIterator()
        }
    }

    override fun hasNext(): Boolean {
        return false
    }

    override fun next(): X {
        throw NoSuchElementException()
    }
}


/**
 * Returns an empty resource iterator.
 */
fun <X> emptyResourceIterator(): ResourceIterator<X> {
    return EmptyResourceIterator()
}

/**
 * Creates a resource iterator that returns the specified values.
 */
fun <X> resourceIteratorOf(vararg elements: X, onClose: () -> Unit = {}): ResourceIterator<X> {
    return WrappedResourceIterator(elements.iterator(), onClose)
}

/**
 * Returns a resource iterator which invokes the function to calculate the next value on each iteration until the function returns `null`.
 * The returned resource iterator is constrained to be iterated only once.
 */
fun <X> generateResourceIterator(next: () -> X?, onClose: () -> Unit = {}): ResourceIterator<X> {
    return GeneratorResourceIterator(next, onClose)
}

/**
 * Returns an [ResourceIterator] that returns the values from the resource iterator.
 * Throws an exception if the resource iterator is constrained to be iterated once and `iterator`
 * is invoked the second time.
 */
fun <X> Sequence<X>.asResourceIterator(onClose: () -> Unit = {}): ResourceIterator<X> {
    return this.iterator().asResourceIterator(onClose)
}

/**
 * Returns an iterator over the elements in this collection.
 * There are no guarantees concerning the order in which the elements are returned
 * (unless this collection is an instance of some class that provides a guarantee).
 */
fun <X> Iterable<X>.asResourceIterator(onClose: () -> Unit = {}): ResourceIterator<X> {
    return this.iterator().asResourceIterator(onClose)
}

/**
 * Returns an [ResourceIterator] that returns the values from the [Iterator].
 */
fun <X> Iterator<X>.asResourceIterator(onClose: () -> Unit = {}): ResourceIterator<X> {
    return WrappedResourceIterator(this, onClose)
}

/**
 * Returns an [ResourceIterator] over the entries in the [Map].
 */
fun <K, V> Map<out K, V>.asResourceIterator(onClose: () -> Unit = {}): ResourceIterator<Map.Entry<K, V>> {
    return this.entries.asResourceIterator(onClose)
}

/**
 * Returns a resource-iterator that yields elements of this resource-iterator sorted according to natural sort order of the value returned by specified [selector] function.
 * The sort is _stable_. It means that equal elements preserve their order relative to each other after sorting.
 * The operation is _intermediate_ and _stateful_.
 */
inline fun <X, R : Comparable<R>> ResourceIterator<X>.sortedBy(crossinline selector: (X) -> R?): ResourceIterator<X> =
    sortedWith(compareBy(selector))

/**
 * Returns a resource-iterator that yields elements of this resource-iterator sorted according to the specified [comparator].
 * The sort is _stable_. It means that equal elements preserve their order relative to each other after sorting.
 * The operation is _intermediate_ and _stateful_.
 */
fun <X> ResourceIterator<X>.sortedWith(comparator: Comparator<in X>): ResourceIterator<X> = use {
    val sortedList = toMutableList()
    sortedList.sortWith(comparator)
    return sortedList.asResourceIterator()
}

/**
 * Returns a resource-iterator that yields elements of this resource-iterator sorted according to their natural sort order.
 * The sort is _stable_. It means that equal elements preserve their order relative to each other after sorting.
 * The operation is _intermediate_ and _stateful_.
 */
fun <T : Comparable<T>> ResourceIterator<T>.sorted(): ResourceIterator<T> = use {
    val sortedList = toMutableList()
    sortedList.sort()
    return sortedList.asResourceIterator()
}

/**
 * Returns a resource-iterator containing only the non-null results of applying the given [transform] function
 * to each element in the original resource-iterator.
 * The operation is _intermediate_ and _stateless_.
 */
@Suppress("UNCHECKED_CAST")
fun <T, R : Any> ResourceIterator<T>.mapNotNull(transform: (T) -> R?): ResourceIterator<R> =
    map(transform).filter { it != null } as ResourceIterator<R>

/**
 * Groups elements of the original resource-iterator by the key returned by the given [keySelector] function
 * applied to each element and returns a map where each group key is associated with a list of corresponding elements.
 * The returned map preserves the entry iteration order of the keys produced from the original resource-iterator.
 * The operation is _terminal_.
 */
inline fun <T, K> ResourceIterator<T>.groupBy(keySelector: (T) -> K): Map<K, List<T>> = use {
    return groupByTo(LinkedHashMap(), keySelector)
}

/**
 * Groups elements of the original resource-iterator by the key returned by the given [keySelector] function
 * applied to each element and puts to the [destination] map each group key associated with a list of corresponding elements.
 * The operation is _terminal_.
 */
inline fun <T, K, M : MutableMap<in K, MutableList<T>>> ResourceIterator<T>.groupByTo(
    destination: M,
    keySelector: (T) -> K,
): M = use {
    for (element in this) {
        val key = keySelector(element)
        val list = destination.getOrPut(key) { arrayListOf() }
        list.add(element)
    }
    return destination
}

/**
 * Returns a [Map] containing the elements from the given resource-iterator indexed by the key
 * returned from [keySelector] function applied to each element.
 * If any two elements had the same key returned by [keySelector] the last one gets added to the map.
 * The returned map preserves the entry iteration order of the original resource-iterator.
 * The operation is _terminal_.
 */
fun <T, K> ResourceIterator<T>.associateBy(keySelector: (T) -> K): Map<K, T> = use {
    asInternalSequence().associateBy(keySelector)
}

/**
 * Returns a [Map] containing key-value pairs provided by [transform] function
 * applied to elements of the given resource-iterator.
 * If any of two pairs have the same key, the last one gets added to the map.
 * The returned map preserves the entry iteration order of the original resource-iterator.
 * The operation is _terminal_.
 */
fun <T, K, V> ResourceIterator<T>.associate(transform: (T) -> Pair<K, V>): Map<K, V> = use {
    asInternalSequence().associate(transform)
}

/**
 * Analogy of the `Sequence<T>.associateByTo`.
 */
inline fun <T, K, M : MutableMap<in K, in T>> ResourceIterator<T>.associateByTo(
    destination: M,
    keySelector: (T) -> K
): M =
    use {
        for (i in this) {
            destination.put(keySelector(i), i)
        }
        return destination
    }

/**
 * Analogy of the `Sequence<T>.associateTo`.
 */
inline fun <T, K, V, M : MutableMap<in K, in V>> ResourceIterator<T>.associateTo(
    destination: M,
    transform: (T) -> Pair<K, V>
): M = use {
    for (i in this) {
        destination += transform(i)
    }
    return destination
}

/**
 * Returns a new map containing all key-value pairs from the given resource-iterator of pairs.
 *
 * The returned map preserves the entry iteration order of the original resource-iterator.
 * If any of two pairs have the same key, the last one gets added to the map.
 */
fun <K, V> ResourceIterator<Pair<K, V>>.toMap(): Map<K, V> = use {
    asInternalSequence().toMap()
}

/**
 * Appends all elements to the given [destination] collection.
 * The operation is _terminal_.
 */
fun <T, C : MutableCollection<in T>> ResourceIterator<T>.toCollection(destination: C): C = use {
    for (item in this) {
        destination.add(item)
    }
    return destination
}

/**
 * Accumulates value starting with [initial] value and applying [operation] from left to right
 * to current accumulator value and each element.
 *
 * Returns the specified [initial] value if the resource-iterator is empty.
 * @param [operation] function that takes current accumulator value and an element, and calculates the next accumulator value.
 * The operation is _terminal_.
 */
inline fun <T, R> ResourceIterator<T>.fold(initial: R, operation: (acc: R, T) -> R): R = use {
    var res = initial
    for (element in this) {
        res = operation(res, element)
    }
    res
}

/**
 * Returns `true` if all elements match the given [predicate].
 * The operation is _terminal_.
 */
inline fun <T> ResourceIterator<T>.all(predicate: (T) -> Boolean): Boolean = use {
    for (element in this) {
        if (!predicate(element)) {
            return false
        }
    }
    return true
}

/**
 * Returns `true` if at least one element matches the given [predicate].
 * The operation is _terminal_.
 */
inline fun <X> ResourceIterator<X>.any(predicate: (X) -> Boolean): Boolean = use {
    for (item in this) {
        if (predicate(item)) {
            return true
        }
    }
    return false
}

/**
 * Performs the given [action] for each remaining element until all elements
 * have been processed or the [action] throws an exception.
 */
inline fun <X> ResourceIterator<X>.forEach(action: (X) -> Unit) {
    use {
        while (hasNext()) {
            action(next())
        }
    }
}

/**
 * Returns a resource-iterator which performs the given [action] on each element of the original resource-iterator as they pass through it.
 * The operation is _intermediate_ and _stateless_.
 */
fun <T> ResourceIterator<T>.onEach(action: (T) -> Unit): ResourceIterator<T> = map {
    action(it)
    it
}


/**
 * Splits this ResourceIterator into a ResourceIterator of lists each not exceeding the given [size].
 * The last list in the resulting ResourceIterator may have fewer elements than the given [size].
 * @param size the number of elements to take in each list,
 * must be positive and can be greater than the number of elements in this sequence.
 * The operation is _intermediate_ and _stateful_.
 */
fun <T> ResourceIterator<T>.chunked(size: Int): ResourceIterator<List<T>> =
    asInternalSequence().chunked(size).asResourceIterator { this.close() }

/**
 * Returns a resource-iterator containing first [n] elements.
 * @throws IllegalArgumentException if [n] is negative or bigger than max-int
 */
fun <T> ResourceIterator<T>.take(n: Long): ResourceIterator<T> {
    require(n >= 0 && n <= Int.MAX_VALUE) { "Requested element count $n should be in range [0, ${Int.MAX_VALUE}]." }
    return asInternalSequence().take(n.toInt()).asResourceIterator {
        this.close()
    }
}

/**
 * Returns a resource-iterator containing all elements except first [n] elements.
 * The operation is _intermediate_ and _stateless_.
 * @throws IllegalArgumentException if [n] is negative or bigger than max-int
 */
fun <T> ResourceIterator<T>.drop(n: Long): ResourceIterator<T> {
    require(n >= 0 && n <= Int.MAX_VALUE) { "Requested element count $n should be in range [0, ${Int.MAX_VALUE}]." }
    return asInternalSequence().drop(n.toInt()).asResourceIterator {
        this.close()
    }
}

/**
 * Safe closes all resources from this [Iterable] by calling [AutoCloseable.close] on each element in turn.
 * It is guaranteed that every [AutoCloseable.close] will be called even if some exception occurs.
 * Note that in case of multiple exception [rootError] is thrown,
 * if there is a single exception it will be thrown, not [rootError].
 */
fun <X> Iterable<X>.closeAll(rootError: Throwable = Exception("Error while closing iterables")) {
    forEach {
        try {
            (it as? AutoCloseable)?.close()
        } catch (ex: Exception) {
            rootError.addSuppressed(ex)
        }
    }
    if (rootError.suppressedExceptions.size == 1) {
        throw rootError.suppressedExceptions[0]
    }
    if (rootError.suppressedExceptions.isNotEmpty()) {
        throw rootError
    }
}

/**
 * Executes safely provided [operations].
 * It is guaranteed that every operation will be executed, even if some exception occurs.
 * Note that in case of multiple exception [rootError] is thrown,
 * if there is a single exception it will be thrown, not rootError.
 */
internal fun safeExec(
    vararg operations: () -> Unit,
    rootError: Throwable = Exception("Error while executing operations")
) {
    operations.forEach {
        try {
            it()
        } catch (ex: Exception) {
            rootError.addSuppressed(ex)
        }
    }
    if (rootError.suppressedExceptions.size == 1) {
        throw rootError.suppressedExceptions[0]
    }
    if (rootError.suppressedExceptions.isNotEmpty()) {
        throw rootError
    }
}

/**
 * Provides an internal [Sequence] to be used for calling terminal Sequence's methods
 * (i.e., those methods that reach the end of the underlying [ResourceIterator]).
 */
private fun <X> ResourceIterator<X>.asInternalSequence(): Sequence<X> {
    require(this is BaseResourceIterator)
    return this.asInternalSequence()
}